package com.gzc.just.play.last.war.centralserver.service.persistence;

import com.gzc.just.play.last.war.centralserver.repository.StatusRepository;
import com.gzc.just.play.last.war.centralserver.user.model.User;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 异步持久化管理器
 * 使用虚拟线程和JUC队列实现写后异步入库
 */
@Service
public class AsyncPersistenceManager {

    private static final Logger logger = LoggerFactory.getLogger(AsyncPersistenceManager.class);

    private final StatusRepository statusRepository;
    private final BlockingQueue<User> userUpdateQueue = new LinkedBlockingQueue<>(10000);
    private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();
    private volatile boolean running = true;

    public AsyncPersistenceManager(StatusRepository statusRepository) {
        this.statusRepository = statusRepository;
    }

    @PostConstruct
    public void start() {
        // start 10 consumers
        for (int i = 0; i < 10; i++) {
            executorService.submit(this::processQueue);
        }
        logger.info("AsyncPersistenceManager started with 10 virtual thread consumers.");
    }

    @PreDestroy
    public void stop() {
        running = false;
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
        }
    }

    /**
     * 提交用户更新任务
     */
    public void submit(User user) {
        if (!userUpdateQueue.offer(user)) {
            logger.warn("User update queue full! Dropping update for user {}", user.getUserId());
            // In production, we might want to fallback to sync write or block, 
            // but for now logging and dropping or throwing exception.
            // Or handle backpressure.
        }
    }

    private void processQueue() {
        while (running && !Thread.currentThread().isInterrupted()) {
            try {
                User user = userUpdateQueue.poll(1, TimeUnit.SECONDS);
                if (user != null) {
                    saveUser(user);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                logger.error("Error processing user update", e);
            }
        }
    }

    private void saveUser(User user) {
        try {
            statusRepository.save(user);
            logger.debug("Async saved user {}", user.getUserId());
        } catch (Exception e) {
            logger.error("Failed to save user {}", user.getUserId(), e);
        }
    }
}
